package ua.com.softway.db.batch;

import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

import javax.persistence.EntityManager;
import javax.persistence.EntityManagerFactory;
import javax.persistence.EntityTransaction;

import org.apache.log4j.Logger;

import ua.com.softway.db.Persistence;

public class BatchProcessor implements Runnable {
    private final static Logger log = Logger.getLogger(BatchProcessor.class);

    @SuppressWarnings({ "rawtypes", "unchecked" })
    private void process(BatchJob job) {
        Batcher batcher = null;
        
        while (true) {
            EntityManagerFactory entityManagerFactory = Persistence.getEntityManagerFactory();
            
            EntityManager entityManager = entityManagerFactory.createEntityManager();
            Persistence.setCurrentEntityManager(entityManager);
            
            EntityTransaction transaction = null;
            
            try {
                transaction = entityManager.getTransaction();
                transaction.begin();
                
                if (batcher == null) batcher = createBatcher(job);
                
                List resultList = batcher.nextBatch(job.createCriteria());
                
                if (resultList == null) return;
                
                job.processBatch(resultList);
                
                if (transaction.isActive()) {
                    entityManager.flush();
                    transaction.commit();
                }
            } catch(RuntimeException e) {
                BatchFailed batchFailed = job.getClass().getAnnotation(BatchFailed.class);
                
                BatchFailedPolicy batchFailedPolicy = batchFailed != null ? batchFailed.value() : BatchFailedPolicy.Abort;
                
                switch (batchFailedPolicy) {
                case Abort:
                    throw e;
                case Ignore:
                    log.error("Batch failed running job " + currentJob + ", execution continued", e);
                    continue;
                }
            } finally {
                if ( transaction != null && transaction.isActive() ) transaction.rollback();

                entityManager.close();
                Persistence.setCurrentEntityManager(null);
            }
        }
    }

    protected Batcher createBatcher(BatchJob<?> job) {
        return new LimitOffsetBatcher(job.getBatchSize());
    }
    
    private void complete(BatchJob<?> job) {
        EntityManagerFactory entityManagerFactory = Persistence.getEntityManagerFactory();
        
        EntityManager entityManager = entityManagerFactory.createEntityManager();
        Persistence.setCurrentEntityManager(entityManager);
        
        EntityTransaction transaction = null;
            
        try {
            transaction = entityManager.getTransaction();
                
            transaction.begin();
            
            job.completed();
                
            if ( transaction.isActive() ) {
                entityManager.flush();
                transaction.commit();
            }
        } finally {
            if ( transaction != null && transaction.isActive() ) transaction.rollback();
                
            entityManager.close();
            Persistence.setCurrentEntityManager(null);
        }
    }

    final BlockingQueue<BatchJob<?>> jobs;
    
    public void addJob(BatchJob<?> job) {
        jobs.add(job);
    }
    
    BatchJob<?> currentJob = null;
    
    @Override
    public void run() {
        while (true) {
            try {
                currentJob = jobs.take();
                
                try {
                    process(currentJob);
                    
                    complete(currentJob);
                } catch(RuntimeException e) {
                    failed(currentJob);
                    
                    log.error("Can't complete job " + currentJob, e);
                } finally {
                    currentJob = null;
                    
                    synchronized (this) {
                        this.notifyAll();
                    }
                }
            } catch (InterruptedException e) {
                return;
            }
        }
    }

    private void failed(BatchJob<?> job) {
        EntityManagerFactory entityManagerFactory = Persistence.getEntityManagerFactory();
        
        EntityManager entityManager = entityManagerFactory.createEntityManager();
        Persistence.setCurrentEntityManager(entityManager);
        
        EntityTransaction transaction = null;
            
        try {
            transaction = entityManager.getTransaction();
                
            transaction.begin();
            
            job.failed();
                
            if ( transaction.isActive() ) {
                entityManager.flush();
                transaction.commit();
            }
        } finally {
            if ( transaction != null && transaction.isActive() ) transaction.rollback();
                
            entityManager.close();
            Persistence.setCurrentEntityManager(null);
        }
    }

    public void waitForCompletion() throws InterruptedException {
        while (isProcessing()) {
            synchronized (this) {
                wait();
            }
        }
    }

    public boolean isProcessing() {
        return currentJob != null || !jobs.isEmpty();
    }
    
    public BatchProcessor(int queueSize) {
        if (instance != null) throw new IllegalStateException();
        instance = this;
        
        jobs = new ArrayBlockingQueue<BatchJob<?>>(queueSize);
    }
    
    public BatchProcessor() {
        this(30);
    }
    
    private static BatchProcessor instance;
    
    public static BatchProcessor instance() {
        if (instance == null) throw new IllegalStateException();
        
        return instance;
    }
}